package com.cl.ks.flow.handler;

import com.alibaba.fastjson.JSONObject;
import com.cl.ks.entity.KsFlowGraph;
import com.cl.ks.entity.SparkNode;
import com.cl.ks.entity.SparkRDDInfo;
import com.cl.ks.flow.base.BaseFlowNodeHandler;
import com.cl.ks.flow.enums.NodeHandlerEnum;
import com.cl.ks.service.SparkRDDInfoService;
import com.cl.ks.service.SysSourceConfigService;
import com.cl.ks.service.SysSourceMappingTableFieldService;
import com.cl.ks.service.SysSourceMappingTableService;
import com.cl.ks.utils.DataSourceMap;
import com.cl.ks.utils.DbUtil;
import com.cl.spark.dto.SparkResult;
import com.querydsl.jpa.impl.JPAQueryFactory;
import org.apache.commons.io.FileUtils;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Optional;


@Component
public class PersistDataHandler extends BaseFlowNodeHandler {
    private final DbUtil dbUtil;
    private final JPAQueryFactory queryFactory;
    private final CreateTableHandler createTableHandler;
    private final PushClueHandler pushClueHandler;
    private final SparkRDDInfoService sparkRDDInfoService;
    @Autowired
    DataSourceMap dataSourceMap;
    @Autowired
    SparkSession sparkSession;

    @Autowired
    private SysSourceConfigService sysSourceConfigService;
    @Autowired
    private SysSourceMappingTableService sysSourceMappingTableService;
    @Autowired
    private SysSourceMappingTableFieldService sysSourceMappingTableFieldService;

    public PersistDataHandler(DbUtil dbUtil, JPAQueryFactory queryFactory, CreateTableHandler createTableHandler, PushClueHandler pushClueHandler, SparkRDDInfoService sparkRDDInfoService) {
        this.dbUtil = dbUtil;
        this.queryFactory = queryFactory;
        this.createTableHandler = createTableHandler;
        this.pushClueHandler = pushClueHandler;
        this.sparkRDDInfoService = sparkRDDInfoService;
    }
    @Override
    public SparkResult process(ProcessParam processParam) {
        KsFlowGraph ksFlowGraph = processParam.getKsFlowGraph();
        SparkNode sparkNode = processParam.getSparkNode();
        Map<String, SparkResult> inputSparkResultList = sparkNode.getInputSparkResultList();
        if (inputSparkResultList.size() > 1) {
            throw new RuntimeException("只能持久化一个节点数据");
        }
        JSONObject nodeExpression = sparkNode.getNodeExpression();
        SparkRDDInfo sparkRDDInfo = new SparkRDDInfo();
        sparkRDDInfo.setName(nodeExpression.getString("name"));
        sparkRDDInfo.setKsFlowGraphId(ksFlowGraph.getId());
        Optional<SparkRDDInfo> exist = sparkRDDInfoService.findByEntity(sparkRDDInfo);
        if (exist.isPresent()) {
            throw new RuntimeException("名称重复");
        }
        sparkRDDInfo.setCode(sparkRDDInfo.getKsFlowGraphId() + ":" + sparkRDDInfo.getName());
        sparkRDDInfo.setType("数据持久化");
        sparkRDDInfo.setNodeCode(sparkNode.getCode());
        SparkResult sparkResult = inputSparkResultList.values().stream().findFirst().get();
        Dataset<Row> persistDataset = sparkResult.getDataset();
        String thisSparkRddDirPath = "./spark-rdd/persist/" + ksFlowGraph.getId() + "/" + sanitizePathSegment(sparkNode.getCode()) + "/";
        File thisSparkRddDir = new File(thisSparkRddDirPath);
        try {
            FileUtils.deleteDirectory(thisSparkRddDir);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        thisSparkRddDir.mkdirs();
        persistDataset.write().mode(SaveMode.Overwrite).parquet(thisSparkRddDirPath);
        sparkRDDInfo.setFilePath(thisSparkRddDirPath);
        sparkRDDInfoService.save(sparkRDDInfo);
        return sparkResult;
    }


    private static String sanitizePathSegment(String pathSegment) {
        // 移除非法字符
        return pathSegment.replaceAll("[^A-Za-z0-9_.-]", "");
    }

    @Override
    public NodeHandlerEnum getType() {
        return NodeHandlerEnum.PERSIST_DATA;
    }
}
